package com.afdor.rws.event.channel;

import com.afdor.rws.event.Event;
import com.afdor.rws.event.channel.client.consumer.EventConsumer;
import com.afdor.rws.event.channel.client.consumer.handler.EventHandler;
import com.afdor.rws.event.channel.client.producer.EventProducer;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.ToString;

import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * 事件传递执行通道抽象实现
 *
 * @author 悭梵
 * @date Created in 2018-08-15 9:31
 */
@ToString
@EqualsAndHashCode
public abstract class AbstractChannelProvider implements ChannelProvider {

    /**
     * 通道上的小芬
     */
    @Getter
    private final Map<String, Queue<EventConsumer>> eventConsumerMap = new ConcurrentHashMap<String, Queue<EventConsumer>>();

    @Override
    public AbstractChannelProvider init() {
        return this;
    }

    @Override
    public void dispose() {
        /**
         * 生成通道关闭
         */
        getEventProducer().destroy();
        /**
         * 消费通道关闭
         */
        eventConsumerMap
                .entrySet()
                .parallelStream()
                .flatMap(entry -> entry.getValue().stream())
                .forEach(item -> item.stop());
    }

    @Override
    public AbstractChannelProvider subscribe(String eventSourceName, EventHandler eventHandler) {
        Queue<EventConsumer> eventConsumerList = eventConsumerMap.get(eventSourceName);
        if (eventConsumerList == null) {
            /* 因生产者、消费者可能都是多线程实现，故采用多线程安全队列 */
            eventConsumerMap.putIfAbsent(eventSourceName, new ConcurrentLinkedQueue<EventConsumer>());
            eventConsumerList = eventConsumerMap.get(eventSourceName);
        }

        EventConsumer eventConsumer = createEventConsumer(eventSourceName, eventHandler);
        eventConsumerList.add(eventConsumer);
        eventConsumer.start();

        return this;
    }

    @Override
    public AbstractChannelProvider sendEvent(Event event) {
        EventProducer eventProducer = getEventProducer();
        eventProducer.sendEvent(event);
        return this;
    }

    /**
     * 由通道具体实现方法来实现具体的消费者创建
     *
     * @param eventSourceName
     * @param eventHandler
     * @return
     */
    protected abstract EventConsumer createEventConsumer(String eventSourceName, EventHandler eventHandler);

    /**
     * 获取事件对应的生产者
     *
     * @return
     */
    protected abstract EventProducer getEventProducer();
}
